{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to XGBoost Spark with GPU\n",
    "\n",
    "Agaricus is an example of xgboost classifier for multiple classification. This notebook will show you how to load data, train the xgboost model.\n",
    "\n",
    "A few libraries required for this notebook:\n",
    "  1. cudf-cu11\n",
    "  2. xgboost\n",
    "  3. scikit-learn\n",
    "  4. numpy\n",
    "  \n",
    "This notebook also illustrates the ease of porting a sample CPU based Spark xgboost4j code into GPU. There is no change required for running Spark XGBoost on GPU because both CPU and GPU call the same API. For CPU run, we need to vectorize the trained dataset before fitting data to classifier."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Import All Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from xgboost.spark import SparkXGBClassifier, SparkXGBClassifierModel\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import FloatType, StructField, StructType\n",
    "from time import time\n",
    "from pyspark.conf import SparkConf\n",
    "import os\n",
    "# if you pass/unpack the archive file and enable the environment\n",
    "# os.environ['PYSPARK_PYTHON'] = \"./environment/bin/python\"\n",
    "# os.environ['PYSPARK_DRIVER_PYTHON'] = \"./environment/bin/python\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Besides CPU version requires two extra libraries.\n",
    "```Python\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.sql.functions import col\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create Spark Session and Data Reader"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-11-30 06:57:40,306 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "2022-11-30 06:57:40,550 WARN resource.ResourceUtils: The configuration of cores (exec = 2 task = 1, runnable tasks = 2) will result in wasted resources due to resource gpu limiting the number of runnable tasks per executor to: 1. Please adjust your configuration.\n",
      "2022-11-30 06:57:54,195 WARN rapids.RapidsPluginUtils: RAPIDS Accelerator 25.02.1 using cudf 25.02.1.\n",
      "2022-11-30 06:57:54,210 WARN rapids.RapidsPluginUtils: spark.rapids.sql.multiThreadedRead.numThreads is set to 20.\n",
      "2022-11-30 06:57:54,214 WARN rapids.RapidsPluginUtils: RAPIDS Accelerator is enabled, to disable GPU support set `spark.rapids.sql.enabled` to false.\n",
      "2022-11-30 06:57:54,214 WARN rapids.RapidsPluginUtils: spark.rapids.sql.explain is set to `NOT_ON_GPU`. Set it to 'NONE' to suppress the diagnostics logging about the query placement on the GPU.\n",
      "2022-11-30 06:57:54,685 WARN yarn.Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.\n"
     ]
    }
   ],
   "source": [
    "SPARK_MASTER_URL = os.getenv(\"SPARK_MASTER_URL\", \"/your-url\")\n",
    "\n",
    "RAPIDS_JAR = os.getenv(\"RAPIDS_JAR\", \"/your-jar-path\")\n",
    "\n",
    "# You need to update with your real hardware resource \n",
    "driverMem = os.getenv(\"DRIVER_MEM\", \"2g\")\n",
    "executorMem = os.getenv(\"EXECUTOR_MEM\", \"2g\")\n",
    "pinnedPoolSize = os.getenv(\"PINNED_POOL_SIZE\", \"2g\")\n",
    "concurrentGpuTasks = os.getenv(\"CONCURRENT_GPU_TASKS\", \"2\")\n",
    "executorCores = int(os.getenv(\"EXECUTOR_CORES\", \"2\"))\n",
    "# Common spark settings\n",
    "conf = SparkConf()\n",
    "conf.setMaster(SPARK_MASTER_URL)\n",
    "conf.setAppName(\"Microbenchmark on GPU\")\n",
    "conf.set(\"spark.executor.instances\",\"1\")\n",
    "conf.set(\"spark.driver.memory\", driverMem)\n",
    "## The tasks will run on GPU memory, so there is no need to set a high host memory\n",
    "conf.set(\"spark.executor.memory\", executorMem)\n",
    "## The tasks will run on GPU cores, so there is no need to use many cpu cores\n",
    "conf.set(\"spark.executor.cores\", executorCores)\n",
    "\n",
    "\n",
    "# Plugin settings\n",
    "conf.set(\"spark.executor.resource.gpu.amount\", \"1\")\n",
    "conf.set(\"spark.rapids.sql.concurrentGpuTasks\", concurrentGpuTasks)\n",
    "conf.set(\"spark.rapids.memory.pinnedPool.size\", pinnedPoolSize)\n",
    "# since pyspark and xgboost share the same GPU, we need to allocate some memory to xgboost to avoid GPU OOM while training \n",
    "conf.set(\"spark.rapids.memory.gpu.allocFraction\",\"0.7\")\n",
    "conf.set(\"spark.locality.wait\",\"0\")\n",
    "##############note: only support value=1 https://github.com/dmlc/xgboost/blame/master/python-package/xgboost/spark/core.py#L370-L374\n",
    "conf.set(\"spark.task.resource.gpu.amount\", 1) \n",
    "conf.set(\"spark.rapids.sql.enabled\", \"true\") \n",
    "conf.set(\"spark.plugins\", \"com.nvidia.spark.SQLPlugin\")\n",
    "conf.set(\"spark.sql.cache.serializer\",\"com.nvidia.spark.ParquetCachedBatchSerializer\")\n",
    "conf.set(\"spark.sql.execution.arrow.maxRecordsPerBatch\", 200000) \n",
    "conf.set(\"spark.driver.extraClassPath\", RAPIDS_JAR)\n",
    "conf.set(\"spark.executor.extraClassPath\", RAPIDS_JAR)\n",
    "\n",
    "# if you pass/unpack the archive file and enable the environment\n",
    "# conf.set(\"spark.yarn.dist.archives\", \"your_pyspark_venv.tar.gz#environment\")\n",
    "# Create spark session\n",
    "spark = SparkSession.builder.config(conf=conf).getOrCreate()\n",
    "\n",
    "reader = spark.read"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Specify the Data Schema and Load the Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "label = 'label'\n",
    "features = [ 'feature_' + str(i) for i in range(0, 126) ]\n",
    "schema = StructType([ StructField(x, FloatType()) for x in [label] + features ])\n",
    "\n",
    "# You need to update them to your real paths!\n",
    "dataRoot = os.getenv(\"DATA_ROOT\", \"/data\")\n",
    "train_path = dataRoot + \"/agaricus/csv/train\"\n",
    "eval_path = dataRoot + \"/agaricus/csv/eval\"\n",
    "\n",
    "data_format = 'csv'\n",
    "has_header = 'true'\n",
    "if data_format == 'csv':\n",
    "    train_data = reader.schema(schema).option('header',has_header).csv(train_path)\n",
    "    trans_data = reader.schema(schema).option('header',has_header).csv(eval_path)\n",
    "else :\n",
    "    train_data = reader.load(train_path)\n",
    "    trans_data = reader.load(eval_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note on CPU version, vectorization is required before fitting data to classifier, which means you need to assemble all feature columns into one column.\n",
    "\n",
    "```Python\n",
    "def vectorize(data_frame):\n",
    "    to_floats = [ col(x.name).cast(FloatType()) for x in data_frame.schema ]\n",
    "    return (VectorAssembler()\n",
    "        .setInputCols(features)\n",
    "        .setOutputCol('features')\n",
    "        .transform(data_frame.select(to_floats))\n",
    "        .select(col('features'), col(label)))\n",
    "\n",
    "train_data = vectorize(train_data)\n",
    "trans_data = vectorize(trans_data)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create a XGBoostClassifier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "params = { \n",
    "    \"tree_method\": \"gpu_hist\",\n",
    "    \"grow_policy\": \"depthwise\",\n",
    "    \"num_workers\": 1,\n",
    "    \"device\": \"cuda\",\n",
    "}\n",
    "params['features_col'] = features\n",
    "params['label_col'] = label\n",
    "    \n",
    "classifier = SparkXGBClassifier(**params)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The parameter `num_workers` should be set to the number of GPUs in Spark cluster for GPU version, while for CPU version it is usually equal to the number of the CPU cores.\n",
    "\n",
    "Concerning the tree method, GPU version only supports `gpu_hist` currently, while `hist` is designed and used here for CPU training.\n",
    "\n",
    "An example of CPU classifier:\n",
    "```\n",
    "classifier = SparkXGBClassifier(\n",
    "  feature_col=features,\n",
    "  label_col=label,  \n",
    "  num_workers=1024,\n",
    "  use_gpu=False,\n",
    ")\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Train the Data with Benchmark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "If features_cols param set, then features_col param is ignored.\n",
      "2022-11-30 07:00:45,526 WARN util.package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n",
      "[Stage 5:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training takes 13.92 seconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r",
      "/data/home/yuanli/work/reviews/pr252/pyspark_venv_20221125/lib/python3.8/site-packages/xgboost/sklearn.py:808: UserWarning: Loading a native XGBoost model with Scikit-Learn interface.\n",
      "  warnings.warn(\"Loading a native XGBoost model with Scikit-Learn interface.\")\n"
     ]
    }
   ],
   "source": [
    "def with_benchmark(phrase, action):\n",
    "    start = time()\n",
    "    result = action()\n",
    "    end = time()\n",
    "    print('{} takes {} seconds'.format(phrase, round(end - start, 2)))\n",
    "    return result\n",
    "model = with_benchmark('Training', lambda: classifier.fit(train_data))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Save and Reload the Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "If features_cols param set, then features_col param is ignored.\n"
     ]
    }
   ],
   "source": [
    "model.write().overwrite().save(dataRoot + '/model/agaricus')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "loaded_model = SparkXGBClassifierModel().load(dataRoot + '/model/agaricus')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Transformation and Show Result Sample"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-11-30 07:01:07,030 WARN rapids.GpuOverrides: \n",
      "!Exec <ProjectExec> cannot run on GPU because not all expressions can be replaced; unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [rawPrediction#798, probability#1062]\n",
      "  @Expression <AttributeReference> label#254 could run on GPU\n",
      "  @Expression <AttributeReference> feature_0#255 could run on GPU\n",
      "  @Expression <AttributeReference> feature_1#256 could run on GPU\n",
      "  @Expression <AttributeReference> feature_2#257 could run on GPU\n",
      "  @Expression <AttributeReference> feature_3#258 could run on GPU\n",
      "  @Expression <AttributeReference> feature_4#259 could run on GPU\n",
      "  @Expression <AttributeReference> feature_5#260 could run on GPU\n",
      "  @Expression <AttributeReference> feature_6#261 could run on GPU\n",
      "  @Expression <AttributeReference> feature_7#262 could run on GPU\n",
      "  @Expression <AttributeReference> feature_8#263 could run on GPU\n",
      "  @Expression <AttributeReference> feature_9#264 could run on GPU\n",
      "  @Expression <AttributeReference> feature_10#265 could run on GPU\n",
      "  @Expression <AttributeReference> feature_11#266 could run on GPU\n",
      "  @Expression <AttributeReference> feature_12#267 could run on GPU\n",
      "  @Expression <AttributeReference> feature_13#268 could run on GPU\n",
      "  @Expression <AttributeReference> feature_14#269 could run on GPU\n",
      "  @Expression <AttributeReference> feature_15#270 could run on GPU\n",
      "  @Expression <AttributeReference> feature_16#271 could run on GPU\n",
      "  @Expression <AttributeReference> feature_17#272 could run on GPU\n",
      "  @Expression <AttributeReference> feature_18#273 could run on GPU\n",
      "  @Expression <AttributeReference> feature_19#274 could run on GPU\n",
      "  @Expression <AttributeReference> feature_20#275 could run on GPU\n",
      "  @Expression <AttributeReference> feature_21#276 could run on GPU\n",
      "  @Expression <AttributeReference> feature_22#277 could run on GPU\n",
      "  @Expression <AttributeReference> feature_23#278 could run on GPU\n",
      "  @Expression <AttributeReference> feature_24#279 could run on GPU\n",
      "  @Expression <AttributeReference> feature_25#280 could run on GPU\n",
      "  @Expression <AttributeReference> feature_26#281 could run on GPU\n",
      "  @Expression <AttributeReference> feature_27#282 could run on GPU\n",
      "  @Expression <AttributeReference> feature_28#283 could run on GPU\n",
      "  @Expression <AttributeReference> feature_29#284 could run on GPU\n",
      "  @Expression <AttributeReference> feature_30#285 could run on GPU\n",
      "  @Expression <AttributeReference> feature_31#286 could run on GPU\n",
      "  @Expression <AttributeReference> feature_32#287 could run on GPU\n",
      "  @Expression <AttributeReference> feature_33#288 could run on GPU\n",
      "  @Expression <AttributeReference> feature_34#289 could run on GPU\n",
      "  @Expression <AttributeReference> feature_35#290 could run on GPU\n",
      "  @Expression <AttributeReference> feature_36#291 could run on GPU\n",
      "  @Expression <AttributeReference> feature_37#292 could run on GPU\n",
      "  @Expression <AttributeReference> feature_38#293 could run on GPU\n",
      "  @Expression <AttributeReference> feature_39#294 could run on GPU\n",
      "  @Expression <AttributeReference> feature_40#295 could run on GPU\n",
      "  @Expression <AttributeReference> feature_41#296 could run on GPU\n",
      "  @Expression <AttributeReference> feature_42#297 could run on GPU\n",
      "  @Expression <AttributeReference> feature_43#298 could run on GPU\n",
      "  @Expression <AttributeReference> feature_44#299 could run on GPU\n",
      "  @Expression <AttributeReference> feature_45#300 could run on GPU\n",
      "  @Expression <AttributeReference> feature_46#301 could run on GPU\n",
      "  @Expression <AttributeReference> feature_47#302 could run on GPU\n",
      "  @Expression <AttributeReference> feature_48#303 could run on GPU\n",
      "  @Expression <AttributeReference> feature_49#304 could run on GPU\n",
      "  @Expression <AttributeReference> feature_50#305 could run on GPU\n",
      "  @Expression <AttributeReference> feature_51#306 could run on GPU\n",
      "  @Expression <AttributeReference> feature_52#307 could run on GPU\n",
      "  @Expression <AttributeReference> feature_53#308 could run on GPU\n",
      "  @Expression <AttributeReference> feature_54#309 could run on GPU\n",
      "  @Expression <AttributeReference> feature_55#310 could run on GPU\n",
      "  @Expression <AttributeReference> feature_56#311 could run on GPU\n",
      "  @Expression <AttributeReference> feature_57#312 could run on GPU\n",
      "  @Expression <AttributeReference> feature_58#313 could run on GPU\n",
      "  @Expression <AttributeReference> feature_59#314 could run on GPU\n",
      "  @Expression <AttributeReference> feature_60#315 could run on GPU\n",
      "  @Expression <AttributeReference> feature_61#316 could run on GPU\n",
      "  @Expression <AttributeReference> feature_62#317 could run on GPU\n",
      "  @Expression <AttributeReference> feature_63#318 could run on GPU\n",
      "  @Expression <AttributeReference> feature_64#319 could run on GPU\n",
      "  @Expression <AttributeReference> feature_65#320 could run on GPU\n",
      "  @Expression <AttributeReference> feature_66#321 could run on GPU\n",
      "  @Expression <AttributeReference> feature_67#322 could run on GPU\n",
      "  @Expression <AttributeReference> feature_68#323 could run on GPU\n",
      "  @Expression <AttributeReference> feature_69#324 could run on GPU\n",
      "  @Expression <AttributeReference> feature_70#325 could run on GPU\n",
      "  @Expression <AttributeReference> feature_71#326 could run on GPU\n",
      "  @Expression <AttributeReference> feature_72#327 could run on GPU\n",
      "  @Expression <AttributeReference> feature_73#328 could run on GPU\n",
      "  @Expression <AttributeReference> feature_74#329 could run on GPU\n",
      "  @Expression <AttributeReference> feature_75#330 could run on GPU\n",
      "  @Expression <AttributeReference> feature_76#331 could run on GPU\n",
      "  @Expression <AttributeReference> feature_77#332 could run on GPU\n",
      "  @Expression <AttributeReference> feature_78#333 could run on GPU\n",
      "  @Expression <AttributeReference> feature_79#334 could run on GPU\n",
      "  @Expression <AttributeReference> feature_80#335 could run on GPU\n",
      "  @Expression <AttributeReference> feature_81#336 could run on GPU\n",
      "  @Expression <AttributeReference> feature_82#337 could run on GPU\n",
      "  @Expression <AttributeReference> feature_83#338 could run on GPU\n",
      "  @Expression <AttributeReference> feature_84#339 could run on GPU\n",
      "  @Expression <AttributeReference> feature_85#340 could run on GPU\n",
      "  @Expression <AttributeReference> feature_86#341 could run on GPU\n",
      "  @Expression <AttributeReference> feature_87#342 could run on GPU\n",
      "  @Expression <AttributeReference> feature_88#343 could run on GPU\n",
      "  @Expression <AttributeReference> feature_89#344 could run on GPU\n",
      "  @Expression <AttributeReference> feature_90#345 could run on GPU\n",
      "  @Expression <AttributeReference> feature_91#346 could run on GPU\n",
      "  @Expression <AttributeReference> feature_92#347 could run on GPU\n",
      "  @Expression <AttributeReference> feature_93#348 could run on GPU\n",
      "  @Expression <AttributeReference> feature_94#349 could run on GPU\n",
      "  @Expression <AttributeReference> feature_95#350 could run on GPU\n",
      "  @Expression <AttributeReference> feature_96#351 could run on GPU\n",
      "  @Expression <AttributeReference> feature_97#352 could run on GPU\n",
      "  @Expression <AttributeReference> feature_98#353 could run on GPU\n",
      "  @Expression <AttributeReference> feature_99#354 could run on GPU\n",
      "  @Expression <AttributeReference> feature_100#355 could run on GPU\n",
      "  @Expression <AttributeReference> feature_101#356 could run on GPU\n",
      "  @Expression <AttributeReference> feature_102#357 could run on GPU\n",
      "  @Expression <AttributeReference> feature_103#358 could run on GPU\n",
      "  @Expression <AttributeReference> feature_104#359 could run on GPU\n",
      "  @Expression <AttributeReference> feature_105#360 could run on GPU\n",
      "  @Expression <AttributeReference> feature_106#361 could run on GPU\n",
      "  @Expression <AttributeReference> feature_107#362 could run on GPU\n",
      "  @Expression <AttributeReference> feature_108#363 could run on GPU\n",
      "  @Expression <AttributeReference> feature_109#364 could run on GPU\n",
      "  @Expression <AttributeReference> feature_110#365 could run on GPU\n",
      "  @Expression <AttributeReference> feature_111#366 could run on GPU\n",
      "  @Expression <AttributeReference> feature_112#367 could run on GPU\n",
      "  @Expression <AttributeReference> feature_113#368 could run on GPU\n",
      "  @Expression <AttributeReference> feature_114#369 could run on GPU\n",
      "  @Expression <AttributeReference> feature_115#370 could run on GPU\n",
      "  @Expression <AttributeReference> feature_116#371 could run on GPU\n",
      "  @Expression <AttributeReference> feature_117#372 could run on GPU\n",
      "  @Expression <AttributeReference> feature_118#373 could run on GPU\n",
      "  @Expression <AttributeReference> feature_119#374 could run on GPU\n",
      "  @Expression <AttributeReference> feature_120#375 could run on GPU\n",
      "  @Expression <AttributeReference> feature_121#376 could run on GPU\n",
      "  @Expression <AttributeReference> feature_122#377 could run on GPU\n",
      "  @Expression <AttributeReference> feature_123#378 could run on GPU\n",
      "  @Expression <AttributeReference> feature_124#379 could run on GPU\n",
      "  @Expression <AttributeReference> feature_125#380 could run on GPU\n",
      "  !Expression <Alias> UDF(pythonUDF0#1327.rawPrediction) AS rawPrediction#798 cannot run on GPU because expression Alias UDF(pythonUDF0#1327.rawPrediction) AS rawPrediction#798 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7; input expression ScalaUDF UDF(pythonUDF0#1327.rawPrediction) (org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 is not supported)\n",
      "    !Expression <ScalaUDF> UDF(pythonUDF0#1327.rawPrediction) cannot run on GPU because neither UDF implemented by class org.apache.spark.ml.functions$$$Lambda$3659/488666387 provides a GPU implementation, nor the conf `spark.rapids.sql.rowBasedUDF.enabled` is enabled; expression ScalaUDF UDF(pythonUDF0#1327.rawPrediction) produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "      @Expression <GetStructField> pythonUDF0#1327.rawPrediction could run on GPU\n",
      "        @Expression <AttributeReference> pythonUDF0#1327 could run on GPU\n",
      "  @Expression <Alias> pythonUDF0#1327.prediction AS prediction#931 could run on GPU\n",
      "    @Expression <GetStructField> pythonUDF0#1327.prediction could run on GPU\n",
      "      @Expression <AttributeReference> pythonUDF0#1327 could run on GPU\n",
      "  !Expression <Alias> UDF(pythonUDF0#1327.probability) AS probability#1062 cannot run on GPU because expression Alias UDF(pythonUDF0#1327.probability) AS probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7; input expression ScalaUDF UDF(pythonUDF0#1327.probability) (org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 is not supported)\n",
      "    !Expression <ScalaUDF> UDF(pythonUDF0#1327.probability) cannot run on GPU because neither UDF implemented by class org.apache.spark.ml.functions$$$Lambda$3659/488666387 provides a GPU implementation, nor the conf `spark.rapids.sql.rowBasedUDF.enabled` is enabled; expression ScalaUDF UDF(pythonUDF0#1327.probability) produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "      @Expression <GetStructField> pythonUDF0#1327.probability could run on GPU\n",
      "        @Expression <AttributeReference> pythonUDF0#1327 could run on GPU\n",
      "\n",
      "2022-11-30 07:01:07,071 WARN rapids.GpuOverrides: \n",
      "!Exec <InMemoryTableScanExec> cannot run on GPU because unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [rawPrediction, probability]; not all expressions can be replaced; unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [rawPrediction#798, probability#1062]\n",
      "  @Expression <AttributeReference> label#254 could run on GPU\n",
      "  @Expression <AttributeReference> feature_0#255 could run on GPU\n",
      "  @Expression <AttributeReference> feature_1#256 could run on GPU\n",
      "  @Expression <AttributeReference> feature_2#257 could run on GPU\n",
      "  @Expression <AttributeReference> feature_3#258 could run on GPU\n",
      "  @Expression <AttributeReference> feature_4#259 could run on GPU\n",
      "  @Expression <AttributeReference> feature_5#260 could run on GPU\n",
      "  @Expression <AttributeReference> feature_6#261 could run on GPU\n",
      "  @Expression <AttributeReference> feature_7#262 could run on GPU\n",
      "  @Expression <AttributeReference> feature_8#263 could run on GPU\n",
      "  @Expression <AttributeReference> feature_9#264 could run on GPU\n",
      "  @Expression <AttributeReference> feature_10#265 could run on GPU\n",
      "  @Expression <AttributeReference> feature_11#266 could run on GPU\n",
      "  @Expression <AttributeReference> feature_12#267 could run on GPU\n",
      "  @Expression <AttributeReference> feature_13#268 could run on GPU\n",
      "  @Expression <AttributeReference> feature_14#269 could run on GPU\n",
      "  @Expression <AttributeReference> feature_15#270 could run on GPU\n",
      "  @Expression <AttributeReference> feature_16#271 could run on GPU\n",
      "  @Expression <AttributeReference> feature_17#272 could run on GPU\n",
      "  @Expression <AttributeReference> feature_18#273 could run on GPU\n",
      "  @Expression <AttributeReference> feature_19#274 could run on GPU\n",
      "  @Expression <AttributeReference> feature_20#275 could run on GPU\n",
      "  @Expression <AttributeReference> feature_21#276 could run on GPU\n",
      "  @Expression <AttributeReference> feature_22#277 could run on GPU\n",
      "  @Expression <AttributeReference> feature_23#278 could run on GPU\n",
      "  @Expression <AttributeReference> feature_24#279 could run on GPU\n",
      "  @Expression <AttributeReference> feature_25#280 could run on GPU\n",
      "  @Expression <AttributeReference> feature_26#281 could run on GPU\n",
      "  @Expression <AttributeReference> feature_27#282 could run on GPU\n",
      "  @Expression <AttributeReference> feature_28#283 could run on GPU\n",
      "  @Expression <AttributeReference> feature_29#284 could run on GPU\n",
      "  @Expression <AttributeReference> feature_30#285 could run on GPU\n",
      "  @Expression <AttributeReference> feature_31#286 could run on GPU\n",
      "  @Expression <AttributeReference> feature_32#287 could run on GPU\n",
      "  @Expression <AttributeReference> feature_33#288 could run on GPU\n",
      "  @Expression <AttributeReference> feature_34#289 could run on GPU\n",
      "  @Expression <AttributeReference> feature_35#290 could run on GPU\n",
      "  @Expression <AttributeReference> feature_36#291 could run on GPU\n",
      "  @Expression <AttributeReference> feature_37#292 could run on GPU\n",
      "  @Expression <AttributeReference> feature_38#293 could run on GPU\n",
      "  @Expression <AttributeReference> feature_39#294 could run on GPU\n",
      "  @Expression <AttributeReference> feature_40#295 could run on GPU\n",
      "  @Expression <AttributeReference> feature_41#296 could run on GPU\n",
      "  @Expression <AttributeReference> feature_42#297 could run on GPU\n",
      "  @Expression <AttributeReference> feature_43#298 could run on GPU\n",
      "  @Expression <AttributeReference> feature_44#299 could run on GPU\n",
      "  @Expression <AttributeReference> feature_45#300 could run on GPU\n",
      "  @Expression <AttributeReference> feature_46#301 could run on GPU\n",
      "  @Expression <AttributeReference> feature_47#302 could run on GPU\n",
      "  @Expression <AttributeReference> feature_48#303 could run on GPU\n",
      "  @Expression <AttributeReference> feature_49#304 could run on GPU\n",
      "  @Expression <AttributeReference> feature_50#305 could run on GPU\n",
      "  @Expression <AttributeReference> feature_51#306 could run on GPU\n",
      "  @Expression <AttributeReference> feature_52#307 could run on GPU\n",
      "  @Expression <AttributeReference> feature_53#308 could run on GPU\n",
      "  @Expression <AttributeReference> feature_54#309 could run on GPU\n",
      "  @Expression <AttributeReference> feature_55#310 could run on GPU\n",
      "  @Expression <AttributeReference> feature_56#311 could run on GPU\n",
      "  @Expression <AttributeReference> feature_57#312 could run on GPU\n",
      "  @Expression <AttributeReference> feature_58#313 could run on GPU\n",
      "  @Expression <AttributeReference> feature_59#314 could run on GPU\n",
      "  @Expression <AttributeReference> feature_60#315 could run on GPU\n",
      "  @Expression <AttributeReference> feature_61#316 could run on GPU\n",
      "  @Expression <AttributeReference> feature_62#317 could run on GPU\n",
      "  @Expression <AttributeReference> feature_63#318 could run on GPU\n",
      "  @Expression <AttributeReference> feature_64#319 could run on GPU\n",
      "  @Expression <AttributeReference> feature_65#320 could run on GPU\n",
      "  @Expression <AttributeReference> feature_66#321 could run on GPU\n",
      "  @Expression <AttributeReference> feature_67#322 could run on GPU\n",
      "  @Expression <AttributeReference> feature_68#323 could run on GPU\n",
      "  @Expression <AttributeReference> feature_69#324 could run on GPU\n",
      "  @Expression <AttributeReference> feature_70#325 could run on GPU\n",
      "  @Expression <AttributeReference> feature_71#326 could run on GPU\n",
      "  @Expression <AttributeReference> feature_72#327 could run on GPU\n",
      "  @Expression <AttributeReference> feature_73#328 could run on GPU\n",
      "  @Expression <AttributeReference> feature_74#329 could run on GPU\n",
      "  @Expression <AttributeReference> feature_75#330 could run on GPU\n",
      "  @Expression <AttributeReference> feature_76#331 could run on GPU\n",
      "  @Expression <AttributeReference> feature_77#332 could run on GPU\n",
      "  @Expression <AttributeReference> feature_78#333 could run on GPU\n",
      "  @Expression <AttributeReference> feature_79#334 could run on GPU\n",
      "  @Expression <AttributeReference> feature_80#335 could run on GPU\n",
      "  @Expression <AttributeReference> feature_81#336 could run on GPU\n",
      "  @Expression <AttributeReference> feature_82#337 could run on GPU\n",
      "  @Expression <AttributeReference> feature_83#338 could run on GPU\n",
      "  @Expression <AttributeReference> feature_84#339 could run on GPU\n",
      "  @Expression <AttributeReference> feature_85#340 could run on GPU\n",
      "  @Expression <AttributeReference> feature_86#341 could run on GPU\n",
      "  @Expression <AttributeReference> feature_87#342 could run on GPU\n",
      "  @Expression <AttributeReference> feature_88#343 could run on GPU\n",
      "  @Expression <AttributeReference> feature_89#344 could run on GPU\n",
      "  @Expression <AttributeReference> feature_90#345 could run on GPU\n",
      "  @Expression <AttributeReference> feature_91#346 could run on GPU\n",
      "  @Expression <AttributeReference> feature_92#347 could run on GPU\n",
      "  @Expression <AttributeReference> feature_93#348 could run on GPU\n",
      "  @Expression <AttributeReference> feature_94#349 could run on GPU\n",
      "  @Expression <AttributeReference> feature_95#350 could run on GPU\n",
      "  @Expression <AttributeReference> feature_96#351 could run on GPU\n",
      "  @Expression <AttributeReference> feature_97#352 could run on GPU\n",
      "  @Expression <AttributeReference> feature_98#353 could run on GPU\n",
      "  @Expression <AttributeReference> feature_99#354 could run on GPU\n",
      "  @Expression <AttributeReference> feature_100#355 could run on GPU\n",
      "  @Expression <AttributeReference> feature_101#356 could run on GPU\n",
      "  @Expression <AttributeReference> feature_102#357 could run on GPU\n",
      "  @Expression <AttributeReference> feature_103#358 could run on GPU\n",
      "  @Expression <AttributeReference> feature_104#359 could run on GPU\n",
      "  @Expression <AttributeReference> feature_105#360 could run on GPU\n",
      "  @Expression <AttributeReference> feature_106#361 could run on GPU\n",
      "  @Expression <AttributeReference> feature_107#362 could run on GPU\n",
      "  @Expression <AttributeReference> feature_108#363 could run on GPU\n",
      "  @Expression <AttributeReference> feature_109#364 could run on GPU\n",
      "  @Expression <AttributeReference> feature_110#365 could run on GPU\n",
      "  @Expression <AttributeReference> feature_111#366 could run on GPU\n",
      "  @Expression <AttributeReference> feature_112#367 could run on GPU\n",
      "  @Expression <AttributeReference> feature_113#368 could run on GPU\n",
      "  @Expression <AttributeReference> feature_114#369 could run on GPU\n",
      "  @Expression <AttributeReference> feature_115#370 could run on GPU\n",
      "  @Expression <AttributeReference> feature_116#371 could run on GPU\n",
      "  @Expression <AttributeReference> feature_117#372 could run on GPU\n",
      "  @Expression <AttributeReference> feature_118#373 could run on GPU\n",
      "  @Expression <AttributeReference> feature_119#374 could run on GPU\n",
      "  @Expression <AttributeReference> feature_120#375 could run on GPU\n",
      "  @Expression <AttributeReference> feature_121#376 could run on GPU\n",
      "  @Expression <AttributeReference> feature_122#377 could run on GPU\n",
      "  @Expression <AttributeReference> feature_123#378 could run on GPU\n",
      "  @Expression <AttributeReference> feature_124#379 could run on GPU\n",
      "  @Expression <AttributeReference> feature_125#380 could run on GPU\n",
      "  !Expression <AttributeReference> rawPrediction#798 cannot run on GPU because expression AttributeReference rawPrediction#798 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "  @Expression <AttributeReference> prediction#931 could run on GPU\n",
      "  !Expression <AttributeReference> probability#1062 cannot run on GPU because expression AttributeReference probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-11-30 07:01:09,857 WARN rapids.GpuOverrides:                               \n",
      "!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it\n",
      "  @Partitioning <SinglePartition$> could run on GPU\n",
      "  !Exec <ProjectExec> cannot run on GPU because unsupported data types in input: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [probability#1062, rawPrediction#798]; not all expressions can be replaced\n",
      "    @Expression <Alias> cast(label#254 as string) AS label#3936 could run on GPU\n",
      "      @Expression <Cast> cast(label#254 as string) could run on GPU\n",
      "        @Expression <AttributeReference> label#254 could run on GPU\n",
      "    @Expression <Alias> cast(rawPrediction#798 as string) AS rawPrediction#3937 could run on GPU\n",
      "      !Expression <Cast> cast(rawPrediction#798 as string) cannot run on GPU because Cast from org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 to StringType is not supported\n",
      "        !Expression <AttributeReference> rawPrediction#798 cannot run on GPU because expression AttributeReference rawPrediction#798 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "    @Expression <Alias> cast(probability#1062 as string) AS probability#3938 could run on GPU\n",
      "      !Expression <Cast> cast(probability#1062 as string) cannot run on GPU because Cast from org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 to StringType is not supported\n",
      "        !Expression <AttributeReference> probability#1062 cannot run on GPU because expression AttributeReference probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "    @Expression <Alias> cast(prediction#931 as string) AS prediction#3939 could run on GPU\n",
      "      @Expression <Cast> cast(prediction#931 as string) could run on GPU\n",
      "        @Expression <AttributeReference> prediction#931 could run on GPU\n",
      "    !Exec <InMemoryTableScanExec> cannot run on GPU because unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [rawPrediction, probability]; unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [probability#1062, rawPrediction#798]; not all expressions can be replaced\n",
      "      @Expression <AttributeReference> label#254 could run on GPU\n",
      "      @Expression <AttributeReference> prediction#931 could run on GPU\n",
      "      !Expression <AttributeReference> probability#1062 cannot run on GPU because expression AttributeReference probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "      !Expression <AttributeReference> rawPrediction#798 cannot run on GPU because expression AttributeReference rawPrediction#798 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Transformation takes 3.26 seconds\n",
      "+-----+--------------------+--------------------+----------+\n",
      "|label|       rawPrediction|         probability|prediction|\n",
      "+-----+--------------------+--------------------+----------+\n",
      "|  1.0|[-9.6646747589111...|[6.35385513305664...|       1.0|\n",
      "|  0.0|[-8.3923015594482...|[2.26557254791259...|       1.0|\n",
      "|  0.0|[-8.0568389892578...|[3.16858291625976...|       1.0|\n",
      "|  0.0|[1.91234850883483...|[0.87128275632858...|       0.0|\n",
      "|  0.0|[-8.5582475662231...|[1.91867351531982...|       1.0|\n",
      "+-----+--------------------+--------------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def transform():\n",
    "    result = loaded_model.transform(trans_data).cache()\n",
    "    result.foreachPartition(lambda _: None)\n",
    "    return result\n",
    "result = with_benchmark('Transformation', transform)\n",
    "result.select(label, 'rawPrediction', 'probability', 'prediction').show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2022-11-30 07:01:10,292 WARN rapids.GpuOverrides: \n",
      "! <DeserializeToObjectExec> cannot run on GPU because not all expressions can be replaced; GPU does not currently support the operator class org.apache.spark.sql.execution.DeserializeToObjectExec\n",
      "  ! <CreateExternalRow> createexternalrow(prediction#931, label#5899, 1.0#5900, newInstance(class org.apache.spark.ml.linalg.VectorUDT).deserialize, StructField(prediction,DoubleType,true), StructField(label,DoubleType,true), StructField(1.0,DoubleType,false), StructField(probability,org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7,true)) cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.catalyst.expressions.objects.CreateExternalRow\n",
      "    @Expression <AttributeReference> prediction#931 could run on GPU\n",
      "    @Expression <AttributeReference> label#5899 could run on GPU\n",
      "    @Expression <AttributeReference> 1.0#5900 could run on GPU\n",
      "    ! <Invoke> newInstance(class org.apache.spark.ml.linalg.VectorUDT).deserialize cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.catalyst.expressions.objects.Invoke\n",
      "      ! <NewInstance> newInstance(class org.apache.spark.ml.linalg.VectorUDT) cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.catalyst.expressions.objects.NewInstance\n",
      "      !Expression <AttributeReference> probability#1062 cannot run on GPU because expression AttributeReference probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "  !Expression <AttributeReference> obj#5905 cannot run on GPU because expression AttributeReference obj#5905 produces an unsupported type ObjectType(interface org.apache.spark.sql.Row)\n",
      "  !Exec <ProjectExec> cannot run on GPU because not all expressions can be replaced; unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [probability#1062]; unsupported data types in input: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [probability#1062]\n",
      "    @Expression <AttributeReference> prediction#931 could run on GPU\n",
      "    @Expression <Alias> cast(label#254 as double) AS label#5899 could run on GPU\n",
      "      @Expression <Cast> cast(label#254 as double) could run on GPU\n",
      "        @Expression <AttributeReference> label#254 could run on GPU\n",
      "    @Expression <Alias> 1.0 AS 1.0#5900 could run on GPU\n",
      "      @Expression <Literal> 1.0 could run on GPU\n",
      "    !Expression <AttributeReference> probability#1062 cannot run on GPU because expression AttributeReference probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "    !Exec <InMemoryTableScanExec> cannot run on GPU because unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [rawPrediction, probability]; not all expressions can be replaced; unsupported data types in output: org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7 [probability#1062]\n",
      "      @Expression <AttributeReference> label#254 could run on GPU\n",
      "      @Expression <AttributeReference> prediction#931 could run on GPU\n",
      "      !Expression <AttributeReference> probability#1062 cannot run on GPU because expression AttributeReference probability#1062 produces an unsupported type org.apache.spark.ml.linalg.VectorUDT@3bfc3ba7\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Evaluation takes 1.0 seconds\n",
      "Accuracy is 0.9069677632722861\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 12:>                                                         (0 + 1) / 1]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "accuracy = with_benchmark(\n",
    "    'Evaluation',\n",
    "    lambda: MulticlassClassificationEvaluator().setLabelCol(label).evaluate(result))\n",
    "print('Accuracy is ' + str(accuracy))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Stop"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
